package es

import (
	"context"
	"fmt"
	"github.com/olivere/elastic/v7"
)

/*
mac 启动 es
https://www.elastic.co/cn/start

*/

type ESClient struct {
	client               *elastic.Client
	KafkaToEsLogDataChan chan interface{}
	// 对应es数据库
	index string
}

var esClient = new(ESClient)

func Init(chansize, num int, urls, userName, passWord, index string) (err error) {

	client, err := elastic.NewClient(elastic.SetURL(urls), elastic.SetBasicAuth(userName, passWord))
	if err != nil {
		return
	}
	//client, err = elastic.NewClient(elastic.SetURL("http://127.0.0.1:9200"), elastic.SetBasicAuth("elastic", "DgbtwgSZJEl=PfAsij5J"))
	esClient.client = client
	esClient.KafkaToEsLogDataChan = make(chan interface{}, chansize)

	// 将kafka发送到通道的数据拿出来发送到es
	for i := 0; i < num; i++ {
		go sendToEs(index, i)
	}

	return
}

func sendToEs(index string, goId int) {
	for msg := range esClient.KafkaToEsLogDataChan {
		fmt.Println("发往es的数据，发送者是goroutine:", goId)
		fmt.Println("数据为：", msg)
		_, err := esClient.client.Index().
			Index(index).
			BodyJson(msg).
			Do(context.Background())
		if err != nil {
			continue
		}
		//fmt.Printf("Indexed user %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)

	}
}

func PutLogData(msg interface{}) {
	esClient.KafkaToEsLogDataChan <- msg

}
